//
// Created by suzhen on 16-11-11.
//

#ifndef OKREDIS_OKREDIS_H
#define OKREDIS_OKREDIS_H

#include <cstdlib>
#include <list>
#include "hiredis/adapters/libuv.h"
#include <OrionV2/Future.hpp>

#include <string>
#include <mutex>

namespace NoSQL {

    struct RawRedisReply {

        RawRedisReply() : RawRedisReply(nullptr) {}

        explicit RawRedisReply(void *reply) : reply_(reply) {};

        operator redisReply *() {
            return (redisReply *) reply_;
        }

        RawRedisReply(RawRedisReply &&o) : reply_(o.reply_) {
        }

        RawRedisReply &operator=(RawRedisReply &&o) {
            if (&o == this) {
                return *this;
            }
            reply_ = o.reply_;
            return *this;
        }

    private:
        void *reply_;
    };

    struct RedisCommandPackage {

    public:

        RedisCommandPackage(const std::string &cmd,
                            std::function<void(RawRedisReply &)> cb
        ) : cmd(cmd), cb(cb), hasInvoked(false) {}

        void invoke(RawRedisReply &r) {
            hasInvoked = true;
            cb(r);
        }

        ~RedisCommandPackage() {
            if (!hasInvoked) {
                RawRedisReply r;
                cb(r);
            }
        }

        const std::string &getCmd() const {
            return cmd;
        }

    private:
        std::string cmd;
        std::function<void(RawRedisReply &)> cb;
        bool hasInvoked;
    };

    class OkRedisCore {

    public:
        using CallBackType = std::function<void(RawRedisReply &)>;

    public:
        static OkRedisCore *createRedisCore(const std::string &ip, int port);

    protected:
        virtual ~OkRedisCore();

        OkRedisCore(const std::string &ip, int port);

    public:
        OkRedisCore(const OkRedisCore &) = delete;

        OkRedisCore &operator=(const OkRedisCore &) = delete;

    private:
        std::thread::id runInThreadId;
        uv_loop_t okRedisEvb_;
        uv_async_t asyncRedisCmdContext_;
        uv_async_t asyncCloseContext_;
        std::list<RedisCommandPackage *> workQueue_;
        std::string ipAddr_;
        int port_;
        redisAsyncContext *redisAsyncCtx;
        std::recursive_mutex commandLock_;
        bool isClosing_;
        Orion::V2::Promise<bool> closePromise_;
        Orion::V2::Promise<bool> startPromise_;
        std::size_t openInHandlers_;
        int refCount_;

    public:

        void retain();

        void release();

        Orion::V2::Future<bool> start();

        Orion::V2::Future<bool> shutdown();

        Orion::V2::Future<bool> closeFuture();

        bool isClosed();

        void executeAsyncCmd(const std::string &cmd,
                             CallBackType &&cb);

    private:
        static void RedisConnected(const redisAsyncContext *ctx, int status);

        static void RedisDisconnected(const redisAsyncContext *ctx, int status);

        static void RedisAsyncCmdCallBack(redisAsyncContext *ctx, void *reply, void *privdata);

        static void OkRedisAsyncOperateCallBack(uv_async_t *async);

        static void OkRedisAsyncCloseCallBack(uv_async_t *async);

        static void OkRedisUninstallHandlersCallBack(uv_handle_t *hd);

    };

}

#endif //OKREDIS_OKREDIS_H
